package com.suning.sawp.service.impl.bi;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import javax.annotation.Resource;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.gson.reflect.TypeToken;
import com.suning.framework.sedis.JedisAction;
import com.suning.framework.sedis.JedisClient;
import com.suning.sawp.constants.BiRealTimeConstants;
import com.suning.sawp.constants.ConfigConstants;
import com.suning.sawp.dto.region.RegionDto;
import com.suning.sawp.intf.bi.BiKafkaRealTimeSaleService;
import com.suning.sawp.intf.dao.BiKafkaRealTimeSaleDaoService;
import com.suning.sawp.intf.store.StoreInfoService;
import com.suning.sawp.po.bi.BiKafkaRealTimeSale;
import com.suning.sawp.po.store.StoreInfo;
import com.suning.sawp.service.util.DJStringUtils;
import com.suning.sawp.service.util.DateUtils;
import com.suning.sawp.service.util.GsonUtils;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;

@Service("biKafkaRealTimeSaleService")
public class BiKafkaRealTimeSaleServiceImpl implements BiKafkaRealTimeSaleService {

    private static final Logger LOGGER = LoggerFactory.getLogger(BiKafkaRealTimeSaleServiceImpl.class);

    /**
     * 门店实时销售前缀
     */
    private static final String STORE_SALE_PREFIX = "com.suning.snsawp.bikafka.storeSale_";

    /**
     * 门店品类实时销售前缀
     */
    private static final String STORE_CATE_SALE_PREFIX = "com.suning.snsawp.bikafka.storeCateSale_";

    private static final String STORE_BRAND_SALE_PREFIX = "com.suning.snsawp.bikafka.storeBrandSale_";
    
    private static final String STORE_YB_CATE_STAFF_PREFIX = "com.suning.snsawp.bikafka.storeYbStaff_";
    
    private static final String STORE_YB_CATE_PREFIX = "com.suning.snsawp.bikafka.storeYbCate_";

    private static final String STAFF_SALE_PREFIX = "com.suning.snsawp.bikafka.staffSale_";

    /**
     * 门店半点实时销售前缀
     */
    private static final String STORE_SALE_HALF_HOUR_PREFIX = "com.suning.snsawp.storeSale.halfHour_";

    /**
     * 门店品类半点实时销售前缀
     */
    private static final String STORE_CATE_SALE_HALF_HOUR_PREFIX = "com.suning.snsawp.storeCateSale.halfHour_";
    
    private Map<String,String> ybMap = new HashMap<String,String>();

    @Resource
    JedisClient jedisClient;

    @Resource
    BiKafkaRealTimeSaleDaoService biKafkaRealTimeSaleDaoService;

    @Resource
    StoreInfoService storeInfoService;
    


    /**
     * 
     * 功能描述: 更新门店实时销售额<br>
     * 〈功能详细描述〉
     * 报文格式：
         {
            "strIdGj": "7611",
            "payAmt": 500,
            "saleCnt": 5.0,
            "payAmt_f": -100,
            "saleCnt_f": -1.0
            "payAmt_r": 600,
            "saleCnt_r": 6.0,
            "memberCount": 2,
            "default_time": "1478880000000",
            "updateTime": "2017-02-10 14:33:36",
            "flag": "snsawp_A002",
            "statisDate": "20170210",
        }
     * @param timeSale
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void updateRedisStoreSale(final Map<String, Object> timeSale) {
        if (null != timeSale && timeSale.containsKey("strIdGj") && null != timeSale.get("strIdGj")) {
            // 门店销售KEY
            final String storeSaleKey = STORE_SALE_PREFIX + String.valueOf(timeSale.get("strIdGj"));
            final long expireAtTime = redisExpireTime();
            jedisClient.execute(new JedisAction<String>() {
                @Override
                public String doAction(Jedis jedis) {
                    Pipeline pipeline = jedis.pipelined();
                    pipeline.set(storeSaleKey, GsonUtils.toJson(timeSale));
                    pipeline.expireAt(storeSaleKey, expireAtTime);
                    pipeline.sync();
                    return null;
                }
            });
        }
    }

    /**
     * 
     * 功能描述: 更新门店品类实时销售额<br>
     * 〈功能详细描述〉
     * 报文格式:
        {
            "deptCd": "00001",
            "strIdGj": "7611",
            "updateTime": "2017-02-10 14:33:36",
            "saleCnt_f": -1.0,
            "payAmt_f": -100,
            "memberCount": 2,
            "saleCnt_r": 6.0,
            "flag": "snsawp_A001",
            "default_time": "1478880000000",
            "payAmt": 500,
            "payAmt_r": 600,
            "saleCnt": 5.0,
            "statisDate": "20170210"
        } 
     * @param timeSale
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void updateRedisStoreCateSale(final Map<String, Object> timeSale) {
        if (null != timeSale && timeSale.containsKey("strIdGj") && timeSale.containsKey("deptCd")
                && null != timeSale.get("strIdGj") && null != timeSale.get("deptCd")) {
            // 门店品类销售KEY
            final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + String.valueOf(timeSale.get("strIdGj"));
            final String cateCode = String.valueOf(timeSale.get("deptCd"));
            final long expireAtTime = redisExpireTime();
            jedisClient.execute(new JedisAction<String>() {
                @Override
                public String doAction(Jedis jedis) {
                    Pipeline pipeline = jedis.pipelined();
                    // 门店品类销售
                    pipeline.hset(storeCateSaleKey, cateCode, GsonUtils.toJson(timeSale));
                    pipeline.expireAt(storeCateSaleKey, expireAtTime);
                    pipeline.sync();
                    return null;
                }
            });
        }
    }

    /**
     * 功能描述: 门店品类品牌的销售相关数据持久化到redis<br>
     * 〈功能详细描述〉
     * 报文格式:
        {
            "deptCd": "00001",
            "strIdGj": "7611",
            "updateTime": "2017-02-10 14:33:36",
            "saleCnt_f": -1.0,
            "payAmt_f": -100,
            "memberCount": 2,
            "brandCd": "000011234",
            "brandName": "小天鹅",
            "default_time": "1478880000000",
            "saleCnt_r": 6.0,
            "flag": "snsawp_A003",
            "payAmt": 500,
            "payAmt_r": 600,
            "saleCnt": 5.0,
            "statisDate": "20170210"
        }
     * @param data
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void updateRedisStoreBrandSale(final Map<String, Object> data) {
        if (!(null != data && data.containsKey("strIdGj") && data.containsKey("brandCd") && null != data.get("strIdGj") && null != data
                .get("brandCd"))) {
            return;
        }
        // redis key
        final String key = STORE_BRAND_SALE_PREFIX + String.valueOf(data.get("strIdGj"));
        // redis hset的field
        final String field = String.valueOf(data.get("brandCd"));

        final long expireAtTime = redisExpireTime();

        jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                // 门店品类品牌销售
                pipeline.hset(key, field, GsonUtils.toJson(data));
                pipeline.expireAt(key, expireAtTime);
                pipeline.sync();
                return null;
            }
        });
    }

    /**
     *     
     * 功能描述: <br>
     * 〈功能详细描述〉
     * {
        strIdGj:'7611',
        salesPerson:'12061818',
        l4GdsGroupCd:'',
        l4GdsGroupDesc:'',
        saleCnt:1.0,
        saleCnt_f:2.0,
        saleCnt_r:-1.0,
        payAmt:200.0,
        payAmt_f:300.0,
        payAmt_r:-100.0,
        memberCount:10,
        flag:'snsawp_A006',
        statisDate:'20170501'
        }
     *
     * @param data
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private void updateRedisYbStaff(final Map<String, Object> data) {
        if (!(null != data && data.containsKey("strIdGj") && data.containsKey("salesPerson") && data.containsKey("l4GdsGroupCd")
                && null != data.get("strIdGj")  && null != data.get("salesPerson") && null != data .get("l4GdsGroupCd"))) {
            return;
        }
        // redis key
        final String key = STORE_YB_CATE_STAFF_PREFIX + String.valueOf(data.get("strIdGj"));
        // redis hset的field
        String ybCode = String.valueOf(data.get("l4GdsGroupCd"));
        if(StringUtils.isBlank(getCateCodeByYbCode(ybCode))){
            LOGGER.error("updateRedisYbStaff cateCode is blank , ybCode :{},storeCode:{}", ybCode , data.get("strIdGj")  );
            return;
        }
        
        final String field = getCateCodeByYbCode(ybCode)+ "_" +  String.valueOf(data.get("salesPerson"));

        final long expireAtTime = redisExpireTime();

        jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                pipeline.hset(key, field, GsonUtils.toJson(data));
                pipeline.expireAt(key, expireAtTime);
                pipeline.sync();
                return null;
            }
        });
    }
    
    private void updateRedisYbCate(final Map<String, Object> data) {
        if (!(null != data && data.containsKey("strIdGj")  && data.containsKey("l4GdsGroupCd")
                && null != data.get("strIdGj") && null != data .get("l4GdsGroupCd"))) {
            return;
        }
        // redis key
        final String key = STORE_YB_CATE_PREFIX + String.valueOf(data.get("strIdGj"));
        // redis hset的field
        String ybCode = String.valueOf(data.get("l4GdsGroupCd"));
        if(StringUtils.isBlank(getCateCodeByYbCode(ybCode))){
            LOGGER.error("updateRedisYbCate cateCode is blank , ybCode :{},storeCode:{}", ybCode , data.get("strIdGj")  );
            return;
        }
        
        final String field = getCateCodeByYbCode(ybCode);
        
        final long expireAtTime = redisExpireTime();

        jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                Pipeline pipeline = jedis.pipelined();
                // 门店品类品牌销售
                pipeline.hset(key, field, GsonUtils.toJson(data));
                pipeline.expireAt(key, expireAtTime);
                pipeline.sync();
                return null;
            }
        });
    }
    
    
    /**
     * 
     *00001 空调  W0000000 
        00002   冰洗  W0000001 
        00003   黑电  W0000002
        00004   数码  W0000003 
        00005   电脑  W0000004 
        00006   通讯  W0000005 
        00007   生活电器    W0000006 
        00013   厨卫  W0000010 
     *
     * @param ybCode
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private String getCateCodeByYbCode(String ybCode){
        
        if(ybMap == null || StringUtils.isBlank(ybCode)){
            return "";
        }
        if( ybMap.isEmpty()){
            ybMap.put("W0000000", "00001");
            ybMap.put("W0000001", "00002");
            ybMap.put("W0000002", "00003");
            ybMap.put("W0000003", "00004");
            ybMap.put("W0000004", "00005");
            ybMap.put("W0000005", "00006");
            ybMap.put("W0000006", "00007");
            ybMap.put("W0000010", "00013");
        }
        return StringUtils.isBlank(ybMap.get(ybCode)) ? "" : ybMap.get(ybCode);
    }
    
    /**
     * 
     * 功能描述: redis失效时间<br>
     * 〈功能详细描述〉
     * 
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private long redisExpireTime() {
        // 取当前时间的小时数和分钟数
        Date tomorrowZeroPoint = DateUtils.getNextDay(DateUtils.getNowZeroPoint(), 1);
        return tomorrowZeroPoint.getTime() / 1000;
    }

    @Override
    public BigDecimal queryStoreDaySale(String storeCode) {
        LOGGER.debug("queryStoreDaySale enter, storeCode={} ", storeCode);
        BigDecimal saleAmount = new BigDecimal("0.00");
        final String storeSaleKey = STORE_SALE_PREFIX + storeCode;
        String saleAmountStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.get(storeSaleKey);
            }
        });

            Map<String, Object> s = trans2Map(saleAmountStr);
            if (null != s && s.containsKey("payAmt") && null != s.get("payAmt")
                    && NumberUtils.isNumber(String.valueOf(s.get("payAmt")))) {
                saleAmount = new BigDecimal(String.valueOf(s.get("payAmt")));
            } else {
            // redis没取到，则查询数据库补充数据到redis
            final Map<String, Object> dbData = biKafkaRealTimeSaleDaoService.querySaleAmountByStoreCd4Alias(storeCode);

            if (null != dbData && !dbData.isEmpty()) {
                saleAmount = new BigDecimal(String.valueOf(dbData.get("payAmt")));
                final long expireAtTime = redisExpireTime();
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        jedis.set(storeSaleKey, JSON.toJSONString(dbData));
                        jedis.expireAt(storeSaleKey, expireAtTime);
                        return null;
                    }
                });
            }
        }

        return saleAmount;
    }

    /**
     * 
     * 功能描述: 转换redis中存储的实时销售数据字符串<br>
     * 〈功能详细描述〉
     *
     * @param saleAmountStr
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private Map<String, Object> trans2Map(String saleAmountStr) {
        if (StringUtils.isNotBlank(saleAmountStr)) {
            try {
                Map<String, Object> s = JSON.parseObject(saleAmountStr);
                return s;
            } catch (Exception e) {
                LOGGER.error("JSON.parseObject trans2Map error, saleAmountStr:{}", saleAmountStr, e);
                return null;
            }
        } else {
            return null;
        }
    }
    
    @Override
    public BigDecimal queryStoreCateDaySale(String storeCode, final String cateCode) {
        LOGGER.debug("queryStoreCateDaySale enter, storeCode={},cateCode={} ", storeCode, cateCode);
        BigDecimal saleAmount = new BigDecimal("0.00");
        final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeCode;
        
        String saleAmountStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                if (StringUtils.isNotBlank(cateCode)) {
                    return jedis.hget(storeCateSaleKey, cateCode);
                } else {
                    return null;
                }
            }
        });

        Map<String, Object> s = trans2Map(saleAmountStr);

        if (null != s && s.containsKey("payAmt") && null != s.get("payAmt")
                && NumberUtils.isNumber(String.valueOf(s.get("payAmt")))) {
            saleAmount = new BigDecimal(String.valueOf(s.get("payAmt")));
        }

        else {
            // redis没取到，则查询数据库补充数据
            final Map<String, Object> dbData = biKafkaRealTimeSaleDaoService.querySaleAmountByStoreCdCateCd4Alias(
                    storeCode, cateCode);
            if (null != dbData && !dbData.isEmpty()) {
                saleAmount = new BigDecimal(String.valueOf(dbData.get("payAmt")));
                final long expireAtTime = redisExpireTime();
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        jedis.hset(storeCateSaleKey, cateCode, JSON.toJSONString(dbData));
                        jedis.expireAt(storeCateSaleKey, expireAtTime);
                        return null;
                    }
                });
            }
        }
        return saleAmount;
    }
    
    /**
     * 门店-品类日销售数量
     */
    @Override
    public Integer queryStoreCateDaySaleVolume(String storeCode, final String cateCode) {
        LOGGER.debug("queryStoreCateDaySaleVolume enter, storeCode={},cateCode={} ", storeCode, cateCode);
        Integer saleVolume = new Integer(0);
        
        if (StringUtils.isBlank(storeCode) || StringUtils.isBlank(cateCode)) {
            LOGGER.error("queryStoreCateDaySaleVolume. storeCode:{}, cateCode:{}", storeCode, cateCode);
            return saleVolume;
        }
        
        final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeCode;

        String saleStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                if (StringUtils.isNotBlank(cateCode)) {
                    return jedis.hget(storeCateSaleKey, cateCode);
                } else {
                    return null;
                }
            }
        });

        Map<String, Object> s = trans2Map(saleStr);

        if (null != s && s.containsKey("saleCnt") && null != s.get("saleCnt")
                && NumberUtils.isNumber(String.valueOf(s.get("saleCnt")))) {
            saleVolume = new BigDecimal(String.valueOf(s.get("saleCnt"))).intValue();
        } else {
            // redis没取到，则查询数据库补充数据
            final Map<String, Object> dbData = biKafkaRealTimeSaleDaoService.querySaleAmountByStoreCdCateCd4Alias(
                    storeCode, cateCode);
            if (null != dbData && !dbData.isEmpty() && null != dbData.get("saleCnt")
                    && NumberUtils.isNumber(String.valueOf(dbData.get("saleCnt")))) {
                saleVolume = new BigDecimal(String.valueOf(dbData.get("saleCnt"))).intValue();
                final long expireAtTime = redisExpireTime();
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        jedis.hset(storeCateSaleKey, cateCode, JSON.toJSONString(dbData));
                        jedis.expireAt(storeCateSaleKey, expireAtTime);
                        return null;
                    }
                });
            }
        }
        return saleVolume;
    }

    @Override
    public BigDecimal queryStaffDaySale(String staffId) {
        BigDecimal saleAmount = new BigDecimal("0.00");
        final String staffSaleKey = STAFF_SALE_PREFIX + ":" + staffId;
        String saleAmountStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.get(staffSaleKey);
            }
        });
        if (NumberUtils.isNumber(saleAmountStr)) {
            saleAmount = new BigDecimal(saleAmountStr);
        }
        return saleAmount;
    }

    @Override
    public void dealKafkaRealTimeSale(List<Map<String, Object>> timeSales) {
        LOGGER.debug("kafka data miss statisflag, params:{}", GsonUtils.toJson(timeSales));
        for (Map<String, Object> timeSale : timeSales) {
            if (null != timeSales && timeSale.containsKey("flag") && null != timeSale.get("flag")) {

                if (BiKafkaRealTimeSale.StatisFlag.STOTE_SALE.equals(String.valueOf(timeSale.get("flag")))) {
                    // 门店销售
                    updateRedisStoreSale(timeSale);
                } else if (BiKafkaRealTimeSale.StatisFlag.STOTE_CATE_SALE.equals(String.valueOf(timeSale.get("flag")))) {
                    // 门店品类销售
                    updateRedisStoreCateSale(timeSale);
                } else if (BiKafkaRealTimeSale.StatisFlag.STOTE_BRAND_SALE.equals(String.valueOf(timeSale.get("flag")))) {
                    // 门店品类品牌销售
                    updateRedisStoreBrandSale(timeSale);
                } else if (BiKafkaRealTimeSale.StatisFlag.STORE_YB_STAFF_SALE.equals(String.valueOf(timeSale.get("flag")))) {
                    // 门店延保人员
                    updateRedisYbStaff(timeSale);
                } else if (BiKafkaRealTimeSale.StatisFlag.STORE_YB_CATE_SALE.equals(String.valueOf(timeSale.get("flag")))) {
                    // 门店延保品类
                    updateRedisYbCate(timeSale);
                }
            } else {
                LOGGER.warn("kafka data miss statisflag, params:{}", timeSale.toString());
            }
        }
    }

    @Override
    public BigDecimal queryStoreDaySaleYoy(String storeCode) {
        // final String key = STORE_SALE_HALF_HOUR_PREFIX + storeCode;
        final String key = STORE_SALE_PREFIX + storeCode;
        // 取半点时间的门店销售数据
        String saleAmountStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.get(key);
            }
        });

        Map<String, Object> s = trans2Map(saleAmountStr);

        if (null != s && s.containsKey("payAmt") && null != s.get("payAmt")
                && NumberUtils.isNumber(String.valueOf(s.get("payAmt")))) {
            // 查询去年同期的销售数据
            BigDecimal lastYearSaleAmount = biKafkaRealTimeSaleDaoService.queryLastYearStoreSaleAmount(storeCode,
                    getNowHourMin());
            return DJStringUtils.calculateYoy(new BigDecimal(String.valueOf(s.get("payAmt"))), lastYearSaleAmount);
        }
        return null;
    }

    /**
     * 
     * 功能描述: 取当前时间的时分<br>
     * 〈功能详细描述〉
     * 
     * @return 格式 HHMM
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private static String getNowHourMin() {
        Calendar cal = Calendar.getInstance();
        cal.setTime(new Date());
        int hour = cal.get(Calendar.HOUR_OF_DAY);
        int min = cal.get(Calendar.MINUTE);
        StringBuilder builder = new StringBuilder();
        builder.append(hour).append(min < 10 ? "0" + min : min);
        return builder.toString();
    }
    
    /**
     *获取半整点分钟数
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private static String getNowHourMinEx() {
        Calendar cal = Calendar.getInstance();
        cal.setTime(new Date());
        int hour = cal.get(Calendar.HOUR_OF_DAY);
        int min = cal.get(Calendar.MINUTE);
        StringBuilder builder = new StringBuilder();
        String minStr = "";
        if(min >= 30){
            minStr = "30";
        }
        if(min < 30 ){
            minStr = "00";
        }
        builder.append(hour).append(minStr);
        return builder.toString();
    }

    @Override
    public BigDecimal queryStoreCateDaySaleYoy(String storeCode, final String cateCode) {
        // final String key = STORE_CATE_SALE_HALF_HOUR_PREFIX + storeCode;
        final String key = STORE_CATE_SALE_PREFIX + storeCode;
        // 取半点时间的门店销售数据
        String saleAmountStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                if (StringUtils.isNotBlank(cateCode)) {
                    return jedis.hget(key, cateCode);
                } else {
                    return null;
                }
            }
        });

        Map<String, Object> s = trans2Map(saleAmountStr);

        if (null != s && s.containsKey("payAmt") && null != s.get("payAmt")
                && NumberUtils.isNumber(String.valueOf(s.get("payAmt")))) {

            // 查询去年同期的销售数据
            BigDecimal lastYearSaleAmount = biKafkaRealTimeSaleDaoService.queryLastYearStoreCateSaleAmount(storeCode,
                    cateCode, getNowHourMin());
            return DJStringUtils.calculateYoy(new BigDecimal(String.valueOf(s.get("payAmt"))), lastYearSaleAmount);
        }

        return null;
    }

    @Override
    public void perisitCacheSaleData() {
        // 取所有大区信息
        List<RegionDto> allRegions = storeInfoService.queryAllRegions();
        for (RegionDto dto : allRegions) {
            // 门店日实时销售列表
            List<Map<String, Object>> realTimeSaleDays = new ArrayList<Map<String, Object>>();
            // 门店品类日实时销售列表
            List<Map<String, Object>> realTimeCateSaleDays = new ArrayList<Map<String, Object>>();
            // 取大区下对应的门店
            List<StoreInfo> storeList = storeInfoService.queryStoresByRegionCode(dto.getRegionCode());
            for (StoreInfo storeInfo : storeList) {
                // 门店销售key
                final String storeSaleKey = STORE_SALE_PREFIX + storeInfo.getStoreCode();
                final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeInfo.getStoreCode();
                // 获取redis中该门店的所有数据，放到一个整体的list中返回
                List<Map<String, String>> storeSales = jedisClient
                        .execute(new JedisAction<List<Map<String, String>>>() {
                            @Override
                            public List<Map<String, String>> doAction(Jedis jedis) {
                                List<Map<String, String>> list = new ArrayList<Map<String, String>>();
                                Map<String, Response<String>> map = new HashMap<String, Response<String>>();
                                // 批量读取
                                Pipeline pipeline = jedis.pipelined();
                                // 1.取门店
                                Response<String> storeSaleAmount = pipeline.get(storeSaleKey);
                                map.put("store", storeSaleAmount);
                                // 2.取门店品类
                                for (String cateCode : ConfigConstants.DEFAULT_CATE_MAP.keySet()) {
                                    Response<String> cateSaleAmount = pipeline.hget(storeCateSaleKey, cateCode);
                                    map.put(cateCode, cateSaleAmount);
                                }
                                // 同步并关闭pipeline
                                pipeline.sync();
                                // 读取pipeline返回的Response数据，转换数据结构
                                Map<String, String> tempMap = new HashMap<String, String>();
                                Set<Entry<String, Response<String>>> entries = map.entrySet();
                                for (Entry<String, Response<String>> entry : entries) {
                                    String value = entry.getValue().get();
                                    if (StringUtils.isBlank(value))
                                        continue;
                                    tempMap.put(entry.getKey(), value);
                                }

                                // 返回一个list list的每一个元素是一个门店的相关数据，包括门店销售和门店品类销售
                                list.add(tempMap);
                                return list;
                            }
                        });

                // 将整体的数据list分到对应逻辑的子list中realTimeSaleDays，realTimeCateSaleDays
                for (Map<String, String> store : storeSales) {
                    Set<Entry<String, String>> entries = store.entrySet();
                    for (Entry<String, String> entry : entries) {
                        if ("store".equals(entry.getKey())) {
                            // 门店销售，key为store
                            Map<String, Object> map = trans2Map(entry.getValue());
                            if (null != map) {
                                realTimeSaleDays.add(map);
                            }
                        } else {
                            // 门店品类销售，key为各个品类id(如：00001)
                            Map<String, Object> map = trans2Map(entry.getValue());
                            if (null != map) {
                                realTimeCateSaleDays.add(map);
                            }
                        }
                    }
                }
            }

            // 批量入库
            // 门店实时销售
            biKafkaRealTimeSaleDaoService.batchUpdateStoreRealTimeSale(realTimeSaleDays);
            // 门店品类实时销售
            biKafkaRealTimeSaleDaoService.batchUpdateStoreCateRealTimeSale(realTimeCateSaleDays);
        }

    }

    @Override
    public void copyHalfHourCacheSaleData() {
        boolean flag = true;
        if (flag) {
            // 变更为不从这个key里取，从每日实时数据中取 与离线数据的半点数据做比较 忽略半小时的数据误差
            return;
        }
        // 取所有大区信息
        List<RegionDto> allRegions = storeInfoService.queryAllRegions();
        for (RegionDto dto : allRegions) {
            // 取大区下对应的门店
            List<StoreInfo> storeList = storeInfoService.queryStoresByRegionCode(dto.getRegionCode());
            for (StoreInfo storeInfo : storeList) {
                // 门店销售key
                final String storeSaleKey = STORE_SALE_PREFIX + storeInfo.getStoreCode();
                // 门店品类销售key
                final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeInfo.getStoreCode();
                // 门店半点销售key
                final String storeSaleHalfHourKey = STORE_SALE_HALF_HOUR_PREFIX + storeInfo.getStoreCode();
                // 门店品类半点销售key
                final String storeCateSaleHalfHourKey = STORE_CATE_SALE_HALF_HOUR_PREFIX + storeInfo.getStoreCode();
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        Map<String, Response<String>> map = new HashMap<String, Response<String>>();
                        // 批量读取
                        Pipeline pipeline = jedis.pipelined();
                        // 备份门店销售
                        Response<String> storeSaleAmount = pipeline.get(storeSaleKey);
                        map.put("store", storeSaleAmount);
                        // 备份门店品类销售
                        for (String cateCode : ConfigConstants.DEFAULT_CATE_MAP.keySet()) {
                            Response<String> cateSaleAmount = pipeline.hget(storeCateSaleKey, cateCode);
                            map.put(cateCode, cateSaleAmount);
                        }
                        pipeline.sync();
                        // 批量写入
                        pipeline = jedis.pipelined();
                        Set<Entry<String, Response<String>>> entries = map.entrySet();
                        for (Entry<String, Response<String>> entry : entries) {
                            if ("store".equals(entry.getKey())) {
                                if (StringUtils.isNotBlank(entry.getValue().get())) {
                                    pipeline.set(storeSaleHalfHourKey, entry.getValue().get());
                                }
                            } else {
                                if (StringUtils.isNotBlank(entry.getValue().get())) {
                                    pipeline.hset(storeCateSaleHalfHourKey, entry.getKey(), entry.getValue().get());
                                }
                            }
                        }
                        pipeline.expireAt(storeSaleHalfHourKey, redisExpireTime());
                        pipeline.expireAt(storeCateSaleHalfHourKey, redisExpireTime());
                        pipeline.sync();
                        return null;
                    }
                });
            }
        }
    }

    /**
     * 查询去年同期销售金额
     */
    @Override
    public BigDecimal queryLastYearStoreSaleAmount(String storeCode) {
        // 查询去年同期的销售数据
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreSaleAmount(storeCode, getNowHourMin());
    }
    
    /**
     * 查询去年同期销售情况
     */
    @Override
    public Map<String, Object> queryLastYearStoreSaleData(String storeCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreSaleData(storeCode, getNowHourMin());
    }
	

    
    /**
     * 查询去年同期销售金额
     */
    @Override
    public Map<String, Object> queryLastYearStoreSaleAmountEx(String storeCode) {
        // 查询去年同期的销售数据
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreSaleAmountEx(storeCode, getNowHourMin());
    }

    /**
     * 查询门店实时数据
     */
    @Override
    public Map<String, Object> getStoreSaleDataFromRedis(String storeCode) {
        // 门店销售KEY
        final String storeSaleKey = STORE_SALE_PREFIX + storeCode;
        String storeSale = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.get(storeSaleKey);
            }
        });
        // 如果门店数据不为空时
        if (StringUtils.isNotBlank(storeSale)) {
            // 返回转换对象
            return GsonUtils.fromJson(storeSale, new TypeToken<Map<String, Object>>() {
            });
        } else {
            // redis没取到，则查询数据库补充数据到redis
            final Map<String, Object> dbData = biKafkaRealTimeSaleDaoService.querySaleAmountByStoreCd4Alias(storeCode);
            if (null != dbData && !dbData.isEmpty()) {
                final long expireAtTime = redisExpireTime();
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        jedis.set(storeSaleKey, JSON.toJSONString(dbData));
                        jedis.expireAt(storeSaleKey, expireAtTime);
                        return null;
                    }
                });
                return dbData;
            }
        }
        return null;
    }

    /**
     * 查询品类实时数据
     */
    @Override
    public List<Map<String, Object>> getStoreCateSaleDataFromRedis(String storeCode) {
        // 门店品类销售KEY
        final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeCode;
        // 获取所有品类数据
        Map<String, String> cateSale = jedisClient.execute(new JedisAction<Map<String, String>>() {
            @Override
            public Map<String, String> doAction(Jedis jedis) {
                return jedis.hgetAll(storeCateSaleKey);
            }
        });
        // 品类销售不为空时
        if (null != cateSale && !cateSale.isEmpty()) {
            List<Map<String, Object>> cateSales = new ArrayList<Map<String, Object>>();
            Map<String, Object> cateMap;
            // 迭代品类数组
            for (Map.Entry<String,String> cate : cateSale.entrySet()) {
                cateMap = GsonUtils.fromJson(cate.getValue(), new TypeToken<Map<String, Object>>() {
                });
                // 品类的数据转换成MAP对象
                if (null != cateMap) {
                    cateSales.add(cateMap);
                }
            }
            return cateSales;
        } else {
            // redis没取到，则查询数据库补充数据
            final List<Map<String, Object>> dbData = biKafkaRealTimeSaleDaoService.queryCatesSaleAmountByStoreCdAlias(storeCode);
            if (CollectionUtils.isNotEmpty(dbData)) {
                final long expireAtTime = redisExpireTime();
                for (final Map<String, Object> cate : dbData) {
                    final String cateCode = String.valueOf(cate.get("deptCd"));
                    jedisClient.execute(new JedisAction<String>() {
                        @Override
                        public String doAction(Jedis jedis) {
                            jedis.hset(storeCateSaleKey, cateCode, JSON.toJSONString(cate));
                            jedis.expireAt(storeCateSaleKey, expireAtTime);
                            return null;
                        }
                    });
                }
                return dbData;
            }
        }
        return Collections.emptyList();
    }

    /**
     * 查询品牌实时数据
     */
    @Override
    public List<Map<String, Object>> getStoreBrandSaleDataFromRedis(String storeCode) {
        // 门店品牌销售KEY
        final String storeBrandSalekey = STORE_BRAND_SALE_PREFIX + storeCode;
        // 获取所有品类数据
        Map<String, String> brandSale = jedisClient.execute(new JedisAction<Map<String, String>>() {
            @Override
            public Map<String, String> doAction(Jedis jedis) {
                return jedis.hgetAll(storeBrandSalekey);
            }
        });
        // 品牌销售不为空时
        if (null != brandSale && !brandSale.isEmpty()) {
            List<Map<String, Object>> brandSales = new ArrayList<Map<String, Object>>();
            Map<String, Object> brandMap;
            // 迭代品牌数组
            for (Map.Entry<String,String> brand : brandSale.entrySet()) {
                brandMap = GsonUtils.fromJson(brand.getValue(), new TypeToken<Map<String, Object>>() {
                });
                if (null != brandMap) {
                    brandSales.add(brandMap);
                }
            }
            return brandSales;
        }
        return Collections.emptyList();
    }

    /**
     * 查询同期销量
     */
    @Override
    public BigDecimal queryLastYearStoreSaleVolume(String storeCode, String cateCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreCateSaleVolume(storeCode, cateCode, getNowHourMin());
    }

    /**
     * 查询去年同期客单价
     */
    @Override
    public BigDecimal queryLastYearStoreCustAvgPrice(String storeCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreCustAvgPrice(storeCode, getNowHourMin());
    }
    
    @Override
    public Map<String, Object> getStoreCateSaleDataFromRedis(String storeCode, final String cateCode) {
        Map<String, Object> map = null;
        // 门店品类销售KEY
        final String storeCateSaleKey = STORE_CATE_SALE_PREFIX + storeCode;
        // 获取所有品类数据
        String cateSale = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.hget(storeCateSaleKey, cateCode);
            }
        });
        
        if (StringUtils.isNotBlank(cateSale)) {
            map = GsonUtils.fromJson(cateSale, new TypeToken<Map<String, Object>>() {});
        } else {
            // redis中无值，查询db补偿数据，再入到redis中
            Map<String, Object> dataMap = biKafkaRealTimeSaleDaoService.querySaleAmountByStoreCdCateCd4Alias(storeCode, cateCode);
            if (null != dataMap) {
                final long expireAtTime = redisExpireTime();
                final String json = JSON.toJSONString(dataMap);
                jedisClient.execute(new JedisAction<String>() {
                    @Override
                    public String doAction(Jedis jedis) {
                        jedis.hset(storeCateSaleKey, cateCode, json);
                        jedis.expireAt(storeCateSaleKey, expireAtTime);
                        return null;
                    }
                });
                return dataMap;
            }
        }
        return map;
    }

    @Override
    public BigDecimal getStoreCateSaleAmountYoy(String storeCode, String cateCode) {
        // 取今日的销售
        Map<String, Object> dataMap = getStoreCateSaleDataFromRedis(storeCode, cateCode);
        BigDecimal todaySaleAmount = new BigDecimal(0);
        if (null != dataMap && null != dataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT)) {
            Double todaySaleAmountStr = (Double) dataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT);
            todaySaleAmount = new BigDecimal(todaySaleAmountStr);
        }
        BigDecimal lastYearSaleAmount = biKafkaRealTimeSaleDaoService.queryLastYearStoreCateSaleAmount(
                storeCode, cateCode, getNowHourMin());
        return DJStringUtils.calculateYoyPercent(todaySaleAmount, lastYearSaleAmount);
    }

    @Override
    public BigDecimal getStoreCateCustAvgPriceYoy(String storeCode, String cateCode) {
        // 取今日的销售
        Map<String, Object> dataMap = getStoreCateSaleDataFromRedis(storeCode, cateCode);
        // 今日销售额
        BigDecimal todaySaleAmount = new BigDecimal(0);
        if (null != dataMap && null != dataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT)){
            Double todaySaleAmountStr = (Double) dataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT);
            todaySaleAmount = new BigDecimal(todaySaleAmountStr);
        }
        // 今日会员数
        BigDecimal todayCustNum = new BigDecimal(0);
        if (null != dataMap && null != dataMap.get(BiRealTimeConstants.StoreCateAttributes.CUST_NUM)) {
            Double todayCustNumStr = (Double) dataMap.get(BiRealTimeConstants.StoreCateAttributes.CUST_NUM);
            todayCustNum = new BigDecimal(todayCustNumStr);
        }
        // 今日客单价
        BigDecimal todayCustAvgPrice = new BigDecimal("0.00");
        if (!todayCustNum.equals(new BigDecimal(0))) {
            todayCustAvgPrice = todaySaleAmount.divide(todayCustNum, 2, BigDecimal.ROUND_DOWN);
        }
        // 去年同期的销售数据
        Map<String, Object> lastDataMap = biKafkaRealTimeSaleDaoService.queryLastYearSaleByStoreCdCateCd(
                storeCode, cateCode, getNowHourMin());
        // 去年同期客单价
        BigDecimal lastYearSaleAmount = new BigDecimal(0);
        if (null != lastDataMap && null != lastDataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT)) {
            lastYearSaleAmount = (BigDecimal) lastDataMap.get(BiRealTimeConstants.StoreCateAttributes.SALE_AMOUNT);
        }
        // 去年同期会员数
        BigDecimal lastYearCustNum = new BigDecimal(0);
        if (null != lastDataMap && null != lastDataMap.get(BiRealTimeConstants.StoreCateAttributes.CUST_NUM)) {
            Integer lastYearCustNumStr = (Integer) lastDataMap.get(BiRealTimeConstants.StoreCateAttributes.CUST_NUM);
            lastYearCustNum = new BigDecimal(lastYearCustNumStr);
        }
        // 去年同期客单价
        BigDecimal lastYearCustAvgPrice = new BigDecimal(0);
        if (!lastYearCustNum.equals(new BigDecimal(0))) {
            lastYearCustAvgPrice = lastYearSaleAmount.divide(lastYearCustNum, 2, BigDecimal.ROUND_DOWN);
        }
        
        return DJStringUtils.calculateYoyPercent(todayCustAvgPrice, lastYearCustAvgPrice);
    }
    
    /**
     * 查询门店品类同期销售额
     */
    @Override
    public BigDecimal queryLastYearStoreCateSaleAmount(String storeCode, String cateCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreCateSaleAmount(
                storeCode, cateCode, getNowHourMin());
    }
    
    /**
     *查询门店品类同期销售额
     */
    @Override
    public    Map<String, Object> queryLastYearStoreCateSaleAmountEx(String storeCode, String cateCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreCateSaleAmountEx(
                storeCode, cateCode, getNowHourMin());
    }


    /**
     * 根据店铺和品类编码获取品牌列表REDIS数据，如果cateCode为空则返回门店所有品牌
     */
    @Override
    public List<Map<String, Object>> getStoreBrandSaleDataByCateFromRedis(String storeCode, String cateCode) {
        List<Map<String, Object>> cateBrandList = Lists.newArrayList();
        List<Map<String, Object>> allBrandList = getStoreBrandSaleDataFromRedis(storeCode);
        if(StringUtils.isBlank(cateCode)){
            return allBrandList;
        }
        if(CollectionUtils.isNotEmpty(allBrandList)){
            for(Map<String, Object> item : allBrandList){
                if(cateCode.equals(String.valueOf(item.get("deptCd"))) ){
                    cateBrandList.add(item);
                }
            }
        }
        return cateBrandList;
    }

    @Override
    public  List<Map<String, Object>> getStoreYbCateDataFromRedis(String storeCode) {

        // 门店品牌销售KEY
        final String storeYbCateKey = STORE_YB_CATE_PREFIX + storeCode;
        // 获取所有延保品类数据
        Map<String, String> ybCateSale = jedisClient.execute(new JedisAction<Map<String, String>>() {
            @Override
            public Map<String, String> doAction(Jedis jedis) {
                return jedis.hgetAll(storeYbCateKey);
            }
        });
        // 延保品类销售不为空时
        if (null != ybCateSale && !ybCateSale.isEmpty()) {
            List<Map<String, Object>> ybSales = new ArrayList<Map<String, Object>>();
            Map<String, Object> ybMap;
            for (Map.Entry<String,String> ybInfo : ybCateSale.entrySet()) {
                ybMap = GsonUtils.fromJson(ybInfo.getValue(), new TypeToken<Map<String, Object>>() {
                });
                if (null != ybMap) {
                    ybSales.add(ybMap);
                }
            }
            return ybSales;
        }
        return Collections.emptyList();
    }
    
    
    /**
     * 获取门店延保销售额
     *
     * @param storeCode
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    @Override
    public BigDecimal getStoreYbSaleFromRedis(String storeCode) {
        List<Map<String, Object>> ybSales  = getStoreYbCateDataFromRedis(storeCode);
        BigDecimal totalAmount = new BigDecimal(0);
        if(CollectionUtils.isNotEmpty(ybSales)){
            for(Map<String, Object> item : ybSales){
                BigDecimal saleAmount;
                if (null != item && null != item.get("payAmt")) {
                    saleAmount = DJStringUtils.bigDecimal(String.valueOf(item.get("payAmt")));
                } else {
                    saleAmount = BigDecimal.ZERO;
                }
                //校验对应的延保品类编码编号是不是八大品类
                if(null == item || null == item.get("l4GdsGroupCd")){
                    continue;
                }else{
                    String ybCode = String.valueOf(item.get("l4GdsGroupCd"));
                    if(StringUtils.isBlank( getCateCodeByYbCode(ybCode))){
                        continue;
                    }
                }
                totalAmount = totalAmount.add(saleAmount);
            }
        }
        return  totalAmount;
    }

    /**
     * 获取某品牌的延保销售额
     */
    @Override
    public BigDecimal getStoreYbCateSaleFromRedis(String storeCode, String cateCode) {
        BigDecimal cateAmount = new BigDecimal(0);
        List<Map<String, Object>> ybSales  = getStoreYbCateDataFromRedis(storeCode);
        if(CollectionUtils.isNotEmpty(ybSales)){
            for(Map<String, Object> item : ybSales){
                //校验对应的延保品类编码编号是不是八大品类
                if(null == item || null == item.get("l4GdsGroupCd")){
                    continue;
                }else{
                    String ybCode = String.valueOf(item.get("l4GdsGroupCd"));
                    if(StringUtils.isBlank( getCateCodeByYbCode(ybCode))){
                        continue;
                    }else if( StringUtils.isNotBlank(cateCode) && cateCode.equals(getCateCodeByYbCode(ybCode))){
                        BigDecimal saleAmount;
                        if (null != item && null != item.get("payAmt")) {
                            saleAmount = DJStringUtils.bigDecimal(String.valueOf(item.get("payAmt")));
                        } else {
                            saleAmount = BigDecimal.ZERO;
                        }
                        cateAmount = saleAmount;
                        break;
                    }
                }
            }
        }
        return  cateAmount;
        
    }
    
    /**
     * 某品类下的延保员工列表
     */
    @Override
    public  List<Map<String, Object>> getStoreYbStaffDataFromRedis(String storeCode,String cateCode) {

        // 门店品牌销售KEY
        final String storeYbCateKey = STORE_YB_CATE_STAFF_PREFIX + storeCode;
        // 获取所有延保员工数据
        Map<String, String> ybStaffSale = jedisClient.execute(new JedisAction<Map<String, String>>() {
            @Override
            public Map<String, String> doAction(Jedis jedis) {
                return jedis.hgetAll(storeYbCateKey);
            }
        });
        // 延保品类销售不为空时
        if (null != ybStaffSale && !ybStaffSale.isEmpty()) {
            List<Map<String, Object>> ybSales = new ArrayList<Map<String, Object>>();
            for (Map.Entry<String,String> ybInfo : ybStaffSale.entrySet()) {
                Map<String, Object>  ybMap = GsonUtils.fromJson(ybInfo.getValue(), new TypeToken<Map<String, Object>>() {
                });
                if (null != ybMap) {
                    //校验对应的延保品类编码编号是不是八大品类
                    if( null == ybMap.get("l4GdsGroupCd")){
                        continue;
                    }else{
                        String ybCode = String.valueOf(ybMap.get("l4GdsGroupCd"));
                        if(StringUtils.isBlank( getCateCodeByYbCode(ybCode))){
                            continue;
                        }else if( StringUtils.isNotBlank(cateCode) && cateCode.equals(getCateCodeByYbCode(ybCode))){
                            ybSales.add(ybMap);
                        }
                    }
                }
            }
            return ybSales;
        }
        return Collections.emptyList();
    }

    @Override
    public Map<String, Object> getStoreYbStaffDataFromRedis(String storeCode, String cateCode, String staffId) {

        // 门店品牌销售KEY
        final String storeYbCateKey = STORE_YB_CATE_STAFF_PREFIX + storeCode;
        // 获取所有延保员工数据
        
        final String field = cateCode +"_"+staffId;
        String ybStaffSaleStr = jedisClient.execute(new JedisAction<String>() {
            @Override
            public String doAction(Jedis jedis) {
                return jedis.hget(storeYbCateKey,field);
            }
        });
        
        try {
            Map<String, Object> map = GsonUtils.fromJson(ybStaffSaleStr, new TypeToken<Map<String, Object>>() {});
            return map;
        } catch (Exception e) {
            LOGGER.error("getStoreYbStaffDataFromRedis exption for storeCode:" +storeCode+",cateCode:"+cateCode+",staffId:"+staffId,e);
        }
        return null;
    }

    @Override
    public List<Map<String, Object>> queryLastYearStoreSaleCateList(String storeCode) {
        return biKafkaRealTimeSaleDaoService.queryLastYearStoreSaleCateList(
                storeCode, getNowHourMinEx());
    }
}
